package com.fanxuankai.kafka.connect.debezium.converters;

import cn.hutool.core.date.DateUtil;
import io.debezium.spi.converter.CustomConverter;
import io.debezium.spi.converter.RelationalColumn;
import org.apache.kafka.connect.data.SchemaBuilder;

import java.sql.JDBCType;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.time.ZonedDateTime;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Properties;

/**
 * @author fanxuankai
 */
public class TimestampTzConverter implements CustomConverter<SchemaBuilder, RelationalColumn> {
    private static final List<String> TYPES = Arrays.asList(JDBCType.TIMESTAMP.getName(), "DATETIME");

    @Override
    public void configure(Properties props) {
    }

    @Override
    public void converterFor(RelationalColumn column, ConverterRegistration<SchemaBuilder> registration) {
        for (String type : TYPES) {
            if (!type.equalsIgnoreCase(column.typeName())) {
                continue;
            }
            SchemaBuilder schemaBuilder = org.apache.kafka.connect.data.Timestamp.builder();
            if (column.isOptional()) {
                schemaBuilder = schemaBuilder.optional();
            }
            registration.register(schemaBuilder, value -> {
                if (value == null) {
                    if (column.isOptional()) {
                        return null;
                    } else if (column.hasDefaultValue()) {
                        return column.defaultValue();
                    } else {
                        return null;
                    }
                }
                // toInstant 会自动转成本地时间，轻松解决时区问题
                Timestamp timestamp;
                if (value instanceof LocalDateTime) {
                    LocalDateTime date = (LocalDateTime) value;
                    timestamp = Timestamp.valueOf(date);
                } else if (value instanceof Timestamp) {
                    Timestamp date = (Timestamp) value;
                    timestamp = Timestamp.from(date.toInstant());
                } else if (value instanceof ZonedDateTime) {
                    ZonedDateTime date = (ZonedDateTime) value;
                    timestamp = Timestamp.from(date.toInstant());
                } else if (value instanceof Date) {
                    Date date = (Date) value;
                    timestamp = Timestamp.from(date.toInstant());
                } else if (value instanceof String) {
                    String date = value.toString();
                    timestamp = Timestamp.from(DateUtil.parse(date).toInstant());
                } else {
                    String message = String.format("时间戳转换异常, 字段名: %s, 类型: %s, 值: %s", column.name(), value.getClass()
                            , value);
                    throw new UnsupportedOperationException(message);
                }
                return timestamp;
            });
            break;
        }
    }
}
